#include "redis_topic.hpp"
#include <iostream>
#include <hiredis/hiredis.h>
using namespace std;
Redis::Redis() : publish_context_(nullptr), subscribe_context_(nullptr)
{
}

Redis::~Redis(){
    if(publish_context_ != nullptr){
        redisFree(publish_context_);
    }
    if(subscribe_context_ != nullptr){
        redisFree(subscribe_context_);
    }
}

//连接Redis服务器
bool Redis::connect(){
    publish_context_ = redisConnect("127.0.0.1", 6379);
    if(publish_context_ == nullptr){
        cerr << "connect redis failed!" <<endl;
        return false;
    }
    subscribe_context_ = redisConnect("127.0.0.1", 6379); //连接redis创建上下文
    if(subscribe_context_ == nullptr){
        cerr<< "connect redis failed!" <<endl;
        return false;
    }
    thread t([&](){
        observer_channel_message();
    });
    t.detach();
    cout << "connect redis success!" << endl;
    return true;
}

bool Redis::publish(int channel, string message){
    //相当于
    redisReply *reply = (redisReply *)redisCommand(publish_context_, "PUBLISH %d %s", channel, message.c_str());
    if(reply == nullptr){
        cerr << "publish command failed!" << endl;
        return false;
    }
    
    freeReplyObject(reply);
    return true;
}

bool Redis::subscribe(int channel){
    //redisCommand 会先把命令缓存到context中，然后调用RedisAppendCommand发送给redis
    //redis执行subscribe是阻塞，不会响应，不会给我们一个reply
    if(REDIS_ERR == redisAppendCommand(subscribe_context_, "SUBSCRIBE %d", channel)){
        cerr << "subscribe command failed!" << endl;
        return false;
    }
    int done = 0;
    while(!done){
        if(REDIS_ERR == redisBufferWrite(subscribe_context_, &done)){
            cerr << "subscribe command failed!" << endl;
            return false;
        }
    }
    return true;
}

//取消订阅
bool Redis::unsubscribe(int channel){
    //redisCommand 会先把命令缓存到context中，然后调用RedisAppendCommand发送给redis
    //redis执行subscribe是阻塞，不会响应，不会给我们一个reply
    if(REDIS_ERR == redisAppendCommand(subscribe_context_, "UNSUBSCRIBE %d", channel)){
        cerr << "unsubscribe command failed!" << endl;
        return false;
    }
    
    int done = 0;
    while(!done){
        if(REDIS_ERR == redisBufferWrite(subscribe_context_, &done)){
            cerr << "unsubscribe command failed!" << endl;
            return false;
        }
    }
    return true;
}

//独立线程中接受订阅通道的消息
void Redis::observer_channel_message(){
    redisReply *reply = nullptr;
    while(REDIS_OK == redisGetReply(subscribe_context_, (void**)&reply)){
//  在Redis的发布/订阅模式中，reply返回的消息格式根据消息类型不同而不同。在你的代码中，主要是处理订阅通道的消息，reply包含以下几种类型的消息：

// 订阅确认消息（subscribe）：

// reply->element[0]->str 是 "subscribe" 字符串
// reply->element[1]->str 是被订阅的频道名称
// reply->element[2] 是当前客户端订阅的频道数量
// 收到的实际消息（message）：

// reply->element[0]->str 是 "message" 字符串
// reply->element[1]->str 是来源频道名称
// reply->element[2]->str 是实际的消息内容
// 取消订阅确认消息（unsubscribe）：

// reply->element[0]->str 是 "unsubscribe" 字符串
// reply->element[1]->str 是被取消订阅的频道名称
// reply->element[2] 是当前客户端订阅的频道数量
        if(reply != nullptr && reply->element[1] != nullptr && reply->element[2]->str != nullptr)
        notify_handler_(stoi(reply->element[1]->str), reply->element[2]->str);
        freeReplyObject(reply);
    }
    cerr << "--------------------observer_channel_message 线程退出------------------------" << endl;
}

//初始化业务层上报通道消息的回调对象
void Redis::init_notify_handler(redis_handler handler)
{
    notify_handler_ = handler;
}
